package com.xxpt.common.test;

import com.alibaba.fastjson.JSONObject;
import com.xxpt.common.constant.RabbitMqConstant;
import com.xxpt.common.utils.amqp.util.RabbitMqUtil;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Slf4j
@Component
public class DelayedMqCousumer {
    @Autowired
    RabbitMqUtil rabbitMqUtil;

    /**
     * 消费示例1
     * concurrency 开启几个线程处理
     */
    /*@RabbitListener(queues = RabbitMqConstant.DELAYED_TEST01, concurrency = "2-5")
    public void consumer(String data, Message message, Channel channel) {
        MessageProperties messageProperties = message.getMessageProperties();
        try {
            log.info(new Date().toLocaleString() + "接收到延迟消费信息==>" + data);
            //参数1，消息的序号
            //参数2，是否批量处理.true:将一次性ack所有小于deliveryTag的消息
            channel.basicAck(messageProperties.getDeliveryTag(),false);
        } catch (Exception ex) {
            log.error("DelayedMq.err", ex);
            try {
                //手动确认回滚 拒绝deliveryTag对应的消息，第二个参数是否requeue，true则重新入队列，否则丢弃或者进入死信队列。
                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
            } catch (IOException ioException) {
                ioException.printStackTrace();
            }
        }
    }*/

    /**
     * 消费示例2
     * concurrency 开启几个线程处理
     */
    @RabbitListener(queues = RabbitMqConstant.DELAYED_TEST01, concurrency = "2")
    public void handle(Message message, Channel channel, JSONObject jsonObject) {
        rabbitMqUtil.handleMessageWithRetry(message, channel, jsonObject, RabbitMqConstant.DELAYED_TEST01, this::handleMessage);
    }
    private void handleMessage(JSONObject jsonObject) {
        log.info("接收到消息==>{}", jsonObject.toJSONString());
    }
}
